package com.it.table;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Table api 使用: 连接外部系统
 * @author code1997
 */
public class TableAPIDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(executionEnvironment);
        //source:创建输入表，连接外部系统读取数据
        tableEnvironment.executeSql("create temporary table input table ... WITH ('connector'=)");
        //sink:创建输出表，连接外部系统输出
        tableEnvironment.executeSql("create temporary table output table ... WITH ('connector'=)");
        //表转换逻辑：
        //从source表中转换为新表
        Table inputTable = tableEnvironment.from("inputTable").select();
        //tableEnvironment.fromDataStream(eventSource,$("`timestamp`").as("ts"));
        inputTable.executeInsert("outputTable");


    }
}
